327a19abf7cc8e420a293214848dccc109b6012e,cdap-app-templates/cdap-etl/cdap-etl-batch/src/main/java/co/cask/cdap/templates/etl/batch/ETLBatchTemplate.java,ETLBatchTemplate,validateAdapter,#ETLStage#ETLStage#List#,140

Before Change


    if (transformList.size() == 0) {
      // No transforms. Check only source and sink.
      if (!(TypeToken.of(batchSink.getKeyType()).isAssignableFrom(batchSource.getKeyType()) &&
        TypeToken.of(batchSink.getValueType()).isAssignableFrom(batchSource.getValueType()))) {
        throw new IllegalArgumentException(String.format("Source %s and Sink %s Types don't match",
                                                         source.getName(), sink.getName()));
      }
    } else {
      // Check the first and last transform with source and sink.
      ETLStage firstStage = Iterables.getFirst(transformList, null);
      ETLStage lastStage = Iterables.getLast(transformList);
      Transform firstTransform = Iterables.getFirst(transforms, null);
      Transform lastTransform = Iterables.getLast(transforms);

      if (!(TypeToken.of(firstTransform.getKeyInType()).isAssignableFrom(batchSource.getKeyType()) &&
        TypeToken.of(firstTransform.getValueInType()).isAssignableFrom(batchSource.getValueType()))) {
        throw new IllegalArgumentException(String.format("Source %s and Transform %s Types don't match",
                                                         source.getName(), firstStage.getName()));
      }

After Change


      Transform lastTransform = Iterables.getLast(transforms);

      if (!(isAssignable(batchSource.getKeyType(), firstTransform.getKeyInType()) &&
        (isAssignable(batchSource.getValueType(), firstTransform.getValueInType())))) {
        throw new IllegalArgumentException(String.format("Source %s and Transform %s Types don't match",
                                                         source.getName(), firstStage.getName()));
      }